package org.example.day20250105.flink.jdbc;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.day20250105.ExcelDataHandle;
import org.example.day20250105.domain.OdsXwDetailData;
import org.example.day20250105.flink.DorisConstant;

import java.util.List;

/**
 * INSERT INTO `test_db_sl`.`ods_xw_detail_data` (`id`, `cycle_id`, `step_id`, `step_type`, `step_time`, `total_time`, `detail_current`, `detail_voltage`, `capacity`, `specific_capacity`, `charge_capacity`, `charge_specific_capacity`, `discharge_capacity`, `discharge_specific_capacity`, `energy`, `specific_energy`, `charge_energy`, `charge_specific_energy`, `discharge_energy`, `discharge_specific_energy`, `absolute_time`, `power`, `dqdv`, `dqmdv`, `contactresistance`, `moduleswitch`, `voltage_1`, `voltage_2`, `temperature_1`, `temperature_2`, `aux_voltage_difference`, `aux_temperature_difference`, `file_name`) VALUES ('10019', '1', '12', NULL, '01:11:25', '14:43:40', '1.3318', '3.35', '1.5815', '1581512.33', '1.5815', '1581512.33', '0', '0', '5.1878', '5187842.67', '5.1878', '5187842.67', '0', '0', '2022-12-31 07:28:01', '4.4615', '0', '0', '2515.62', '关闭', '-0.0003', NULL, '25.3', NULL, '0', '0', NULL);
 */
public class OdsXwDetailDataWriteByFlink {
    private static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


    public static void save(List<OdsXwDetailData> odsXwDetailDataList){
        env.setParallelism(1);
        env.fromData(odsXwDetailDataList)
                .addSink(JdbcSink.sink(
                        "INSERT INTO `test_db_sl`.`ods_xw_detail_data` (`id`, `cycle_id`, `step_id`, `step_type`, `step_time`, `total_time`, `detail_current`, `detail_voltage`, `capacity`, " +
                                "`specific_capacity`, `charge_capacity`, `charge_specific_capacity`, `discharge_capacity`, `discharge_specific_capacity`, `energy`, `specific_energy`, `charge_energy`," +
                                " `charge_specific_energy`, `discharge_energy`, `discharge_specific_energy`, `absolute_time`, `power`, `dqdv`, `dqmdv`, `contactresistance`, `moduleswitch`, `voltage_1`, " +
                                "`voltage_2`, `temperature_1`, `temperature_2`, `aux_voltage_difference`, `aux_temperature_difference`, `file_name`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);",
                        (ps, data) -> {
                            ps.setString(1, data.getId());
                            ps.setString(2, data.getCycle_id());
                            ps.setString(3, data.getStep_id());
                            ps.setString(4, data.getStep_type());
                            ps.setString(5, data.getStep_time());
                            ps.setString(6, data.getTotal_time());
                            ps.setString(7, data.getDetail_current());
                            ps.setString(8, data.getDetail_voltage());
                            ps.setString(9, data.getCapacity());
                            ps.setString(10, data.getSpecific_capacity());
                            ps.setString(11, data.getCharge_capacity());
                            ps.setString(12, data.getCharge_specific_capacity());
                            ps.setString(13, data.getDischarge_capacity());
                            ps.setString(14, data.getDischarge_specific_capacity());
                            ps.setString(15, data.getEnergy());
                            ps.setString(16, data.getSpecific_energy());
                            ps.setString(17, data.getCharge_energy());
                            ps.setString(18, data.getCharge_specific_energy());
                            ps.setString(19, data.getDischarge_energy());
                            ps.setString(20, data.getDischarge_specific_energy());
                            ps.setString(21, data.getAbsolute_time());
                            ps.setString(22, data.getPower());
                            ps.setString(23, data.getDqdv());
                            ps.setString(24, data.getDqmdv());
                            ps.setString(25, data.getContactresistance());
                            ps.setString(26, data.getModuleswitch());
                            ps.setString(27, data.getVoltage_1());
                            ps.setString(28, data.getVoltage_2());
                            ps.setString(29, data.getTemperature_1());
                            ps.setString(30, data.getTemperature_2());
                            ps.setString(31, data.getAux_voltage_difference());
                            ps.setString(32, data.getAux_temperature_difference());
                            ps.setString(33, data.getFile_name());
                        },
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUsername(DorisConstant.USERNAME)
                                .withPassword(DorisConstant.PASSWORD)
                                .withDriverName(DorisConstant.DRIVER_CLASS_NAME)
                                .withUrl(DorisConstant.DORIS_JDBC_URL)
                                .build()
                ));

    }

    public static void main(String[] args) throws Exception {
        ExcelDataHandle.readDetailData();
        env.execute();
    }
}
